/**
 * 
 */
package dse.messages.multiversion;

import java.util.*;
import java.io.*;
import dse.*;
import dse.messages.*;
import dse.messages.singleversion.ShareDownloader;

/**
 * @author sam
 *
 */
public class GetQuorumShareTransaction extends Transaction implements
		ShareDownloader, VersionReader {

	//private Node requester;
	private DataObject desiredObject;
	private LinkedList<Node> residentNodes;
	private boolean failed;
	private TreeMap<Version, LinkedList<Node>> sharesDownloaded;
	private int validShareCount;
	private int successCount;
	private int failureCount;
	private int maxSequence;
	private Version maxVersion;
	private int sharesToBeUpdate;
	private boolean hasRequestPreviousShares;
	private Hashtable<TreeSet<Integer>, Integer> quorums;
	private boolean matchOneQuorum;
	private int round;
	
	private static Vector<TreeSet<Integer>> validQuorums;
	static{
		try{
			validQuorums = new Vector<TreeSet<Integer>>();
			Scanner sc = new Scanner(new File("d:\\java\\dse\\quorum.txt"));
			while(sc.hasNextLine()){
				String[] lines = sc.nextLine().split(",");
				TreeSet<Integer> tsi = new TreeSet<Integer>();
				for(int i=0; i<lines.length; i++){
					tsi.add(Integer.parseInt(lines[i].trim()));
				}
				validQuorums.add(tsi);
			}
		}catch(Exception e){
			System.err.println(e.getMessage());
			System.exit(1);
		}
	}
	
	public GetQuorumShareTransaction(Experiment experiment,
			int id,
			Node requester,
			DataObject desiredObject,
			double time
			)
	{
		super(id, requester, experiment, time);
		
		//this.requester = requester;
		this.desiredObject = desiredObject;
		this.failed = false;
		//this.requestShareCount = 0;
		this.validShareCount = 0;
		this.successCount = 0;
		this.failureCount = 0;
		this.maxSequence = -1;
		this.maxVersion = null;
		
		this.sharesToBeUpdate = desiredObject.shareCount();
		if(isUpdate && isPartialUpdate)
		{
			this.sharesToBeUpdate = (int)(desiredObject.shareCount()*experiment.configuration().getRandom().nextDouble());
		}
		this.hasRequestPreviousShares = false;
		this.quorums = new Hashtable<TreeSet<Integer>, Integer>();
		initQuorum();
		this.matchOneQuorum = false;
		this.round = 1;
	}
	
	private void initQuorum(){
		for(TreeSet<Integer> tsi : validQuorums){
			this.quorums.put(tsi, 0);
		}
	}

	@Override
	public void start(ExperimentRecorder recorder) {
		// TODO Auto-generated method stub
		residentNodes = MessageUtil.getNodesByDht(experiment.network(), requester, desiredObject, experiment.modules().assigner());		

		if(isUpdate)
		{
			recorder.dataUpdateInitiated(super.id, requester, desiredObject, startTime);
			recorder.debug(this, "start", startTime);
			//int srf = experiment.modules().shareRequestDecisionMaker().sharesToRequestFor(desiredObject)
				//		+ this.maxFailure*2;
			//this.requestShareCount = desiredObject.shareCount();
			//this.validShareCount = requestShareCount - this.maxFailure;
			
			//for (int i=0; (i<requestShareCount) && (i<residentNodes.size()); i++)
				//sendVersionRequest(recorder, residentNodes.get(i), startTime);
			for (int i=0; i<residentNodes.size(); i++)
				sendVersionRequest(recorder, residentNodes.get(i), startTime);
		}
		else
		{
			recorder.dataRequestInitiated(super.id, requester, desiredObject, startTime);
			recorder.debug(this, "start", startTime);
			
			//this.requestShareCount = experiment.modules().shareRequestDecisionMaker().sharesToRequestFor(desiredObject);
			//this.requestShareCount = this.requestShareCount>this.desiredObject.shareCount()?this.desiredObject.shareCount():this.requestShareCount;
			//this.validShareCount = desiredObject.requiredShareCount();
			for(int i=0; i<residentNodes.size(); i++)
			{
				sendGetLatestShareRequest(recorder, residentNodes.get(i), startTime);
			}		
		}
	}
	
	@Override
	public void shareDownloaded (ExperimentRecorder recorder, Node from,
			DataObject dobj, int shareIndex, double timeLeft)
	{
	}
	
	@Override
	public void shareDownloadFailed (ExperimentRecorder recorder, DataObject dobj, 
			int shareIndex, double timeLeft)
	{		
	}
	
	@Override
	public boolean isDoneDownloadingShares ()
	{
		return (failed || finished);
	}
	
	private void sendGetLatestShareRequest(ExperimentRecorder recorder, Node node, double time)
	{
		if (node == null) {
			// We should turn this into a warning ...
			recorder.warn(
					"Share assigner failed to map share to node in cluster!"
					);
			return;
		}
		
		Message m = super.createRequestLatestShareMessage(requester, node, desiredObject, 
				MessagePurpose.DataObjectRequest, this);
		recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}
	
	private void sendGetPreviousShareRequest(ExperimentRecorder recorder, Version version, 
			Node node, double time)
	{
		if (node == null) {
			// We should turn this into a warning ...
			recorder.warn(
					"Share assigner failed to map share to node in cluster!"
					);
			return;
		}
		
		Message m = super.createRequestPreviousShareMessage(requester, node, desiredObject, 
				maxVersion, MessagePurpose.DataObjectRequest, this, round);
		recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}
	
	@Override
	public void shareUpdated (ExperimentRecorder recorder, Node to,
			DataObject dobj, int shareIndex, double timeLeft)
	{
		if(isUpdate && !this.isDoneUpdatingShares()) {		
			this.successCount++;
			for(TreeSet<Integer> quorum : quorums.keySet())
			{
				if(quorum.contains(shareIndex)){
					quorums.put(quorum, quorums.get(quorum)+1);
					if(quorums.get(quorum) == quorum.size()){
						matchOneQuorum = true;
						break;
					}
				}
			}
			if(matchOneQuorum)
				transactionSucceed(recorder, timeLeft);
			else if((successCount+failureCount) >= this.sharesToBeUpdate)
				transactionFailed(recorder, timeLeft);
		}
	}
	
	@Override
	public void shareUpdateFailed (ExperimentRecorder recorder, DataObject dobj, 
			int shareIndex, double timeLeft)
	{
		if(isUpdate && !this.isDoneUpdatingShares()) {
			this.failureCount++;
			if((successCount + failureCount) >= this.sharesToBeUpdate)
				transactionFailed(recorder, timeLeft);
		}
	}
	
	@Override
	public boolean isDoneUpdatingShares ()
	{
		return (failed || finished);
	}
	
	private void sendUpdateShareRequest(ExperimentRecorder recorder, Node node, Double time)
	{
		if (node == null) {
			// We should turn this into a warning ...
			recorder.warn(
					"Share assigner failed to map share to node in cluster!"
					);
			return;
		}
		
		Message m = super.createShareUpdateV5Message(requester, node, desiredObject, 
				maxVersion, MessagePurpose.DataObjectUpdate, this);
		recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}
	
	@Override
	public void readVersion(ExperimentRecorder recorder, TreeSet<Version> versions, double timeLeft, Node from){}
	
	public void readVersion(ExperimentRecorder recorder, Node from, Version version, double timeLeft)
	{
		if(finished)
			return;
		//if(r < this.round)
			//return;
		if(version == null)
		{
			failureCount++;
			
			if(successCount+failureCount>= desiredObject.shareCount()){
				System.out.println("Failed number exceeds the maximum number");
				transactionFailed(recorder, timeLeft);
			}
			//	processVersion(recorder, timeLeft);
			return;
		}
		if(sharesDownloaded == null)
			sharesDownloaded = new TreeMap<Version, LinkedList<Node>>();
		//if(successCount >= validShareCount)
			//return;
		successCount++;
		if(this.sharesDownloaded.get(version) == null)
		{
			this.sharesDownloaded.put(version, new LinkedList<Node>());
			this.sharesDownloaded.get(version).add(from);
		}
		else
			this.sharesDownloaded.get(version).add(from);
		for(TreeSet<Integer> quorum : quorums.keySet())
		{
			if(quorum.contains(from.getDataShare(desiredObject).getIndex())){
				quorums.put(quorum, quorums.get(quorum)+1);
				if(quorums.get(quorum) == quorum.size()){
					matchOneQuorum = true;
					this.validShareCount = quorum.size();
					break;
				}
			}
		}
		if(matchOneQuorum)
			processVersion(recorder, timeLeft);
		else if(successCount + failureCount >= desiredObject.shareCount()){
			transactionFailed(recorder, timeLeft);
		}
		//	processVersion(recorder, timeLeft);
	}
	
	public void readVersion(ExperimentRecorder recorder, int seq, double timeLeft){}
	public void readVersion(ExperimentRecorder recorder, int seq, int index, double time){
		if(seq == -1)
		{
			failureCount++;
			//if(failureCount>maxFailure && !failed)
			//{
				//System.out.println("Failed number exceeds the maximum number");
				//transactionFailed(recorder, timeLeft);
			//}
			return;
		}
		if(maxSequence < seq)
			maxSequence = seq;
		//if(successCount >= validShareCount)
			//return;
		successCount++;
		for(TreeSet<Integer> quorum : quorums.keySet())
		{
			if(quorum.contains(index)){
				quorums.put(quorum, quorums.get(quorum)+1);
				if(quorums.get(quorum) == quorum.size()){
					matchOneQuorum = true;
					break;
				}
			}
		}
		
		if(matchOneQuorum)
			processVersions(recorder, time);
		else if(successCount+failureCount >= residentNodes.size()){
			transactionFailed(recorder, time);
		}
	}
	
	public void readVersion(ExperimentRecorder recorder, double time, LinkedList<Version> versions, Version version, Node from){}

	private void sendVersionRequest(ExperimentRecorder recorder, Node node, Double time)
	{
		if (node == null) {
			// We should turn this into a warning ...
			recorder.warn(
					"Share assigner failed to map share to node in cluster!"
					);
			return;
		}
		
		Message m = super.createRequestVersionQuorumMessage(
				requester, 
				node,
				desiredObject, 
				MessagePurpose.DataObjectUpdate,
				this
				);
		recorder.debug(this, "new msg: " + m, time);
		experiment.messageQueue().enqueueMessage(m, time);
	}
	
	private void processVersions(ExperimentRecorder recorder, double timeLeft)
	{
		if(!matchOneQuorum || maxSequence<0)
		{
			transactionFailed(recorder, timeLeft);
			return;
		}
		maxVersion = new Version(maxSequence+1, requester.id());
		this.successCount = 0;
		this.failureCount = 0;
		this.matchOneQuorum = false;
		for(TreeSet<Integer> quorum : quorums.keySet())
			quorums.put(quorum, 0);
		
		for(int i=0; i<this.sharesToBeUpdate; i++)
			sendUpdateShareRequest(recorder, residentNodes.get(i), timeLeft);
		//transactionSucceed(recorder, timeLeft);
	}
	
	private void processVersion(ExperimentRecorder recorder, double timeLeft)
	{
		if((sharesDownloaded==null) || (sharesDownloaded.size()==0))
		{
			transactionFailed(recorder, timeLeft);
			return;
		}
		
		maxVersion = sharesDownloaded.lastKey();
		if(maxVersion.getSequence()>requester.getSequence())
			requester.setSequence(maxVersion.getSequence());
		Map.Entry<Version, LinkedList<Node>> vn = sharesDownloaded.pollLastEntry();
		while((vn.getValue().size()<desiredObject.requiredShareCount()) 
				&& (sharesDownloaded.size()>0))
		{
			vn = sharesDownloaded.pollLastEntry();
		}
		
		if(vn.getValue().size() >= this.validShareCount)
		{
			transactionSucceed(recorder, timeLeft);
		}
		else if(vn.getValue().size() < this.validShareCount)
		{
			maxVersion = vn.getKey();
			for(int i=0; i<this.residentNodes.size(); i++){
				if(!vn.getValue().contains(residentNodes.get(i)))
					sendUpdateShareRequest(recorder, residentNodes.get(i), timeLeft);
			}
			transactionSucceed(recorder, timeLeft);	
		}
		else if(sharesDownloaded.size() == 0)
		{	
			//int srf = experiment.modules().shareRequestDecisionMaker().sharesToRequestFor(desiredObject)
				//+ this.maxFailure*2;
			if(this.hasRequestPreviousShares)
				recorder.requestPreviousShares(desiredObject, false);
			else{
				this.hasRequestPreviousShares = true;
				recorder.requestPreviousShares(desiredObject, true);
			}
			failureCount = 0;
			successCount = 0;
			this.matchOneQuorum = false;
			this.round++;
			for(TreeSet<Integer> quorum : quorums.keySet())
				quorums.put(quorum, 0);
			//for(int i=0; (i<srf)&&(i<residentNodes.size()); i++)
			for(int i=0; i<residentNodes.size(); i++)	
				sendGetPreviousShareRequest(recorder, maxVersion, residentNodes.get(i), timeLeft);
		}
		//else
			//transactionFailed(recorder, timeLeft);
	}
	
	private void transactionSucceed(ExperimentRecorder recorder, double timeLeft)
	{
		if(finished)
			return;
		finished = true;
		failed  = false;
		//System.out.println("Data ID : " + desiredObject.id() + "Version : " + maxVersion.getSequence() + " , " + maxVersion.getClientID());
		if(isUpdate)
		{
			recorder.debug(
					"GetDhtDataSharesWithVersionTransaction trans#" + super.id + 
					" is finshed!!!"
					);
			recorder.dataUpdateFinished(super.id, requester, desiredObject, true, timeLeft);
		}
		else
		{
			recorder.debug(
					"GetDhtDataSharesWithVersionTransaction trans#" + super.id + 
					" is finshed!!!"
					);
			recorder.dataRequestFinished(super.id, requester, desiredObject, true, this.residentNodes.get(this.validShareCount-1),
					false, timeLeft);
		}
	}
	
	private void transactionFailed (ExperimentRecorder recorder, double timeLeft)
	{
		if(finished)
			return;
		recorder.debug(this, "transFailed", timeLeft);
		
		finished = true;
		failed = true;
		if(isUpdate)
			recorder.dataUpdateFinished(super.id, requester, desiredObject, false, timeLeft);
		else
			recorder.dataRequestFinished(super.id, requester, desiredObject, false, null, false, timeLeft);
	}
	public Transaction getTransaction(){
		return this;
	}
}
